LangSmith + LangFuse：LLM应用的可观测性全链路追踪

引言

想象一下，你负责的客服机器人突然开始对用户说“抱歉，我无法回答这个问题”。用户投诉如潮水般涌来，你打开日志系统，发现一切看起来正常——HTTP 200，无异常堆栈，响应时间正常。但对话质量就是下降了。

这可能是最让人头疼的调试场景：LLM应用的问题往往不是“系统崩溃”，而是“行为异常”。一次糟糕的Prompt拼接、一个embedding检索结果的微小偏移、一次不恰当的模型温度设置，都可能让输出质量断崖式下跌。

传统APM（如Datadog、SkyWalking）只能看到“服务是否健康”，却无法回答“模型为什么这么回答”。这就是LLM应用可观测性要解决的问题——它不仅追踪系统的“心跳”，还要追踪模型的“思维”。

核心概念

生活类比：餐厅后厨的“透明玻璃”

假设你经营一家餐厅。传统监控就像后厨门口的“温度计”——你只知道厨房没着火（服务存活），但不知道厨师（LLM）为什么把宫保鸡丁做成了糖醋味（输出异常）。

LLM可观测性则是在后厨装了一面透明玻璃，同时记录：

• 顾客点了什么（输入Prompt）

• 厨师看了哪本菜谱（检索到的上下文）

• 厨师放了什么调料、放了多少（模型参数、Token用量）

• 最终端上桌的菜长什么样（输出结果）

• 如果菜品出问题，还能回放当时的做菜过程（Trace回放）

技术定义

LangSmith 和 LangFuse 是两款主流的LLM应用可观测性平台，它们都提供：

• Traces（链路追踪）：记录一次完整请求从进入系统到返回的全过程

• Spans（跨度）：链路中的每个步骤，如Retriever、LLM调用、Tool执行

• Observations（观测点）：每个Span内部的详细数据，如Token用量、延迟、成本

• Evaluations（评估）：对输出质量进行自动化评测

源码/原理深度分析

LangSmith的Trace数据模型

LangSmith的核心数据结构可以简化为以下模型：

# langsmith/schemas.py (简化版)
class TracerSession(BaseModel):
    id: UUID
    name: str
    start_time: datetime
    end_time: Optional[datetime]

class Run(BaseModel):
    id: UUID
    name: str  # 如 "ChatOpenAI"
    run_type: str  # "llm", "chain", "tool", "retriever"
    inputs: dict  # Prompt内容
    outputs: dict  # 模型输出
    start_time: datetime
    end_time: datetime
    extra: dict  # 包含token用量、模型名等
    error: Optional[str]
    parent_run_id: Optional[UUID]  # 构建树形结构
    child_runs: List["Run"] = []

关键设计在于 parent_run_id —— 它构建了一棵多叉树，而非简单的调用栈。这意味着LangSmith可以完美表示LangChain中复杂的DAG（有向无环图）执行流程。

LangFuse的追踪范式

LangFuse采用了OpenTelemetry风格的四层模型：

# langfuse/model.py (简化版)
class Trace(BaseModel):
    id: str
    timestamp: datetime
    name: str
    input: Any
    output: Any
    metadata: dict
    session_id: Optional[str]
    user_id: Optional[str]

class Observation(BaseModel):
    id: str
    trace_id: str
    parent_observation_id: Optional[str]
    type: str  # "SPAN", "GENERATION", "EVENT"
    start_time: datetime
    end_time: Optional[datetime]
    model: Optional[str]  # 对于GENERATION类型
    input: Any
    output: Any
    level: str  # "DEFAULT", "WARNING", "ERROR"
    status_message: Optional[str]
    version: Optional[str]

LangFuse的一个亮点是 type 字段区分了 SPAN（普通步骤）和 GENERATION（模型生成），这让你能快速筛选出所有LLM调用，而不必遍历整棵Trace树。

核心差异：隐式追踪 vs 显式追踪

LangSmith通过 @traceable 装饰器或LangChain回调实现隐式追踪：

from langsmith import traceable
from langchain_openai import ChatOpenAI

@traceable(run_type="chain", name="customer_support")
def handle_query(query: str):
    llm = ChatOpenAI(model="gpt-4o")
    # 内部LLM调用会被自动追踪
    response = llm.invoke(query)
    return response.content

LangFuse则需要显式创建观测点：

from langfuse import Langfuse
from langfuse.model import CreateGeneration

langfuse = Langfuse()

def handle_query(query: str):
    generation = langfuse.generation(
        name="customer-support-llm",
        model="gpt-4o",
        input=query,
        model_parameters={"temperature": 0.7}
    )
    response = call_llm(query)
    generation.end(output=response)
    return response

架构对比图：

graph TD subgraph "LangSmith 隐式追踪" A[应用代码] --> B[@traceable 装饰器] B --> C[LangChain 回调机制] C --> D[自动捕获所有子步骤] D --> E[LangSmith 后端] end subgraph "LangFuse 显式追踪" F[应用代码] --> G[手动创建 Observation] G --> H[手动结束 Observation] H --> I[LangFuse 后端] end subgraph "共同能力" E --> J[Trace 可视化] I --> J J --> K[性能分析] J --> L[质量评估] J --> M[成本追踪] end

实战代码

示例一：LangSmith + LangGraph构建RAG链路追踪

"""
完整可运行的RAG链路追踪示例
前置条件: pip install langsmith langgraph langchain-openai chromadb
环境变量: LANGCHAIN_TRACING_V2=true LANGCHAIN_API_KEY=your_key
"""
import os
from typing import List

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict

# 启用LangSmith追踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "rag-tracing-demo"

class AgentState(TypedDict):
    question: str
    documents: List[str]
    answer: str

def load_and_index_document(file_path: str):
    """加载文档并构建向量索引"""
    loader = TextLoader(file_path)
    documents = loader.load()
    
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50
    )
    chunks = text_splitter.split_documents(documents)
    
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=OpenAIEmbeddings()
    )
    return vectorstore

def retrieve(state: AgentState, vectorstore):
    """检索相关文档片段"""
    retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
    docs = retriever.invoke(state["question"])
    return {"documents": [doc.page_content for doc in docs]}

def generate(state: AgentState):
    """基于检索结果生成回答"""
    llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
    
    context = "\n\n".join(state["documents"])
    prompt = f"""基于以下上下文回答问题。如果无法从上下文中找到答案，请明确说明。

上下文：
{context}

问题：{state["question"]}

回答："""
    
    response = llm.invoke(prompt)
    return {"answer": response.content}

def build_graph(vectorstore):
    """构建LangGraph执行图"""
    workflow = StateGraph(AgentState)
    
    # 定义节点
    workflow.add_node("retrieve", lambda state: retrieve(state, vectorstore))
    workflow.add_node("generate", generate)
    
    # 定义边
    workflow.set_entry_point("retrieve")
    workflow.add_edge("retrieve", "generate")
    workflow.add_edge("generate", END)
    
    return workflow.compile()

# 初始化数据
with open("sample_company_policy.txt", "w") as f:
    f.write("""员工年假政策：每位全职员工每年享有15天带薪年假。
年假可以分多次使用，但每次请假需提前3个工作日提交申请。
未经批准的休假将视为旷工。""")

vectorstore = load_and_index_document("sample_company_policy.txt")
app = build_graph(vectorstore)

# 执行查询 - LangSmith会自动追踪整个执行图
result = app.invoke({"question": "我明年可以请几次年假？"})
print(f"回答：{result['answer']}")

示例二：LangFuse自定义评估指标

"""
LangFuse自定义评估器 - 检测幻觉和有害内容
前置条件: pip install langfuse llama_cpp
"""
from langfuse import Langfuse
from langfuse.model import (
    CreateGeneration,
    CreateSpan,
    CreateTrace,
    Dataset,
    DatasetItem,
    EvaluationResult
)
import json

langfuse = Langfuse(
    public_key="pk-...",
    secret_key="sk-...",
    host="https://cloud.langfuse.com"
)

class LLMQualityEvaluator:
    """自定义LLM输出质量评估器"""
    
    def __init__(self):
        self.evaluation_prompt = """你是质量评估专家。请评估以下AI回答的质量。

用户问题：{question}
AI回答：{answer}
参考标准：{ground_truth}

请从以下维度打分（1-10分）：
1. 准确性：回答与参考标准的一致性
2. 完整性：是否遗漏关键信息
3. 安全性：是否包含有害或敏感内容

输出JSON格式：{{"accuracy": 8, "completeness": 7, "safety": 10, "comment": "简要说明"}}"""
    
    def evaluate(self, question: str, answer: str, ground_truth: str) -> dict:
        """执行评估并返回结构化结果"""
        import openai
        client = openai.OpenAI()
        
        prompt = self.evaluation_prompt.format(
            question=question,
            answer=answer,
            ground_truth=ground_truth
        )
        
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )
        
        return json.loads(response.choices[0].message.content)

def trace_with_evaluation():
    """创建带评估的完整追踪"""
    # 创建Trace
    trace = langfuse.trace(
        name="customer-support-query",
        user_id="user_123",
        session_id="session_456",
        metadata={"tier": "premium"}
    )
    
    # 记录LLM调用
    question = "我的订单为什么还没发货？"
    answer = "您的订单正在处理中，预计24小时内发货。"
    ground_truth = "订单因库存问题延迟，已为您升级优先发货。"
    
    generation = trace.generation(
        name="support-llm",
        model="gpt-4o",
        model_parameters={"temperature": 0.7},
        input=question,
        output=answer
    )
    
    # 运行评估器
    evaluator = LLMQualityEvaluator()
    scores = evaluator.evaluate(question, answer, ground_truth)
    
    # 将评估结果关联到Generation
    trace.span(
        name="quality-evaluation",
        input={"question": question, "answer": answer},
        output=scores,
        level="DEFAULT"
    )
    
    # 记录评估结果
    if scores["safety"] < 8:
        trace.update(
            level="WARNING",
            status_message=f"安全评分过低: {scores['safety']}"
        )
    
    # 最终提交
    trace.update(output={"final_answer": answer})
    
    print(f"追踪ID: {trace.id}")
    print(f"评估结果: {scores}")

if __name__ == "__main__":
    trace_with_evaluation()

示例三：双平台集成——LangSmith追踪 + LangFuse评估

"""
同时使用LangSmith和LangFuse的混合方案
LangSmith负责链路追踪，LangFuse负责评估和实验管理
前置条件: pip install langsmith langfuse langchain-openai
"""
import asyncio
from datetime import datetime
from typing import Dict, Any

from langchain_openai import ChatOpenAI
from langsmith import traceable, Client
from langfuse import Langfuse

# 初始化双客户端
langsmith_client = Client()
langfuse = Langfuse()

class HybridObservabilityManager:
    """统一管理两个可观测性平台"""
    
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o-mini")
        self.evaluation_results = []
    
    @traceable(run_type="chain", name="hybrid_rag_chain")
    async def process_query(self, query: str, context: str = "") -> Dict[str, Any]:
        """主处理函数 - 由LangSmith追踪"""
        
        # 在LangFuse中创建对应的Trace
        lf_trace = langfuse.trace(
            name="hybrid-query",
            input=query,
            metadata={"timestamp": datetime.utcnow().isoformat()}
        )
        
        try:
            # 1. 检索阶段
            retrieval_span = lf_trace.span(name="retrieval")
            retrieved_docs = await self._simulate_retrieval(query, context)
            retrieval_span.end(output={"doc_count": len(retrieved_docs)})
            
            # 2. 生成阶段
            generation = lf_trace.generation(
                name="llm-response",
                model="gpt-4o-mini",
                input=query,
                model_parameters={"temperature": 0.5}
            )
            
            response = await self.llm.ainvoke(
                f"基于以下上下文回答问题：\n\n{retrieved_docs}\n\n问题：{query}"
            )
            
            generation.end(output=response.content)
            
            # 3. 质量评估
            eval_results = await self._run_quality_evaluation(query, response.content)
            lf_trace.span(
                name="evaluation",
                output=eval_results,
                level="WARNING" if eval_results["needs_review"] else "DEFAULT"
            )
            
            # 4. 在LangSmith中也记录评估结果
            self._record_langsmith_evaluation(query, response.content, eval_results)
            
            return {
                "answer": response.content,
                "evaluation": eval_results,
                "langfuse_trace_id": lf_trace.id
            }
            
        except Exception as e:
            lf_trace.span(
                name="error",
                level="ERROR",
                status_message=str(e)
            )
            raise
        
        finally:
            lf_trace.update(output={"status": "completed"})
    
    async def _simulate_retrieval(self, query: str, context: str) -> list:
        """模拟文档检索"""
        await asyncio.sleep(0.1)  # 模拟检索延迟
        return [context[i:i+200] for i in range(0, len(context), 200)][:3]
    
    async def _run_quality_evaluation(self, query: str, answer: str) -> dict:
        """运行质量评估"""
        # 简单规则评估
        needs_review = len(answer) < 50 or "抱歉" in answer
        
        return {
            "needs_review": needs_review,
            "answer_length": len(answer),
            "contains_apology": "抱歉" in answer,
            "score": 8.5 if not needs_review else 5.0
        }
    
    def _record_langsmith_evaluation(self, query: str, answer: str, eval_results: dict):
        """将评估结果记录到LangSmith"""
        langsmith_client.create_feedback(
            run_id=None,  # 实际使用时需要传入run_id
            key="quality_score",
            score=eval_results["score"],
            comment=f"Query: {query[:50]}..."
        )

# 使用示例
async def main():
    manager = HybridObservabilityManager()
    
    # 模拟多个用户请求
    queries = [
        "退款流程是什么？",
        "如何修改账户信息？",
        "你们支持哪些支付方式？"
    ]
    
    context = """
    退款政策：用户可在购买后7天内申请退款。
    账户修改：登录后进入设置页面即可修改个人信息。
    支付方式：我们支持信用卡、支付宝和微信支付。
    """
    
    for query in queries:
        result = await manager.process_query(query, context)
        print(f"查询: {query}")
        print(f"回答: {result['answer'][:100]}...")
        print(f"评估: {result['evaluation']}")
        print("-" * 50)
    
    # 提交所有LangFuse数据
    langfuse.flush()

if __name__ == "__main__":
    asyncio.run(main())

方案对比

| 维度 | LangSmith | LangFuse | 自建Otel方案 |

|------|-----------|----------|-------------|

| 接入成本 | 低（装饰器/回调） | 中（需手动创建） | 高（需自行实现） |

| LangChain集成 | 原生支持 | 需适配器 | 手动埋点 |

| 评估功能 | 内置评估器 | 支持自定义评估 | 无 |

| 实验管理 | 弱 | 强（支持Prompt版本对比） | 无 |

| 数据所有权 | 托管服务 | 支持自托管/私有化 | 完全自有 |

| 成本追踪 | 按模型估算 | 精确到Token | 需自行实现 |

| 团队协作 | 支持 | 支持 | 依赖基础设施 |

适用场景建议：

• 快速验证：选择LangSmith，5分钟接入，零代码侵入

• 生产环境：选择LangFuse，数据可控，评估可定制

• 大型企业：自建Otel方案，完全掌控数据流

最佳实践与避坑指南

最佳实践

1. 统一Trace ID贯穿全链路

   # 将Trace ID关联到业务日志
   trace_id = get_current_trace_id()
   logger.info(f"business_event", extra={"trace_id": trace_id})

1. 控制采样率

   # 生产环境建议10%-20%采样率
   LANGSMITH_SAMPLING_RATE=0.2

1. 敏感信息脱敏

   @traceable(process_inputs=lambda x: {"text": mask_pii(x.get("text", ""))})
   def llm_call(text: str):
       ...

常见坑及解决方案

坑1：Token成本爆炸

• 现象：Trace数据消耗大量Token，成本失控

• 解决：设置采样率、限制输入输出长度、使用gzip压缩

坑2：异步任务丢失Trace

• 现象：Celery/Redis队列中的任务没有Trace

• 解决：手动传递Trace上下文

from langsmith.run_helpers import get_current_run_tree

def async_task():
    parent_run = get_current_run_tree()
    # 在子线程/进程中重建上下文
    with parent_run.create_child(name="async_subtask"):
        ...

坑3：评估结果不准确

• 现象：LLM评估器自身质量不稳定

• 解决：混合使用规则评估和LLM评估，设置置信度阈值

总结

LLM应用的可观测性不是锦上添花，而是生产环境必备的基础设施。LangSmith和LangFuse代表了两种不同的设计哲学：LangSmith追求极致的简洁和无侵入，LangFuse则提供更细粒度的控制和更强的评估能力。

我的建议是：

• 如果你的团队已经在用LangChain，从LangSmith开始

• 如果需要私有化部署和自定义评估，选择LangFuse

• 大型系统可以考虑双平台方案，互相补充

延伸思考：可观测性数据本身就是宝贵的训练数据。想象一下，如果你能自动收集所有失败的Trace，将它们转化为微调数据集，或者用它们来优化Prompt模板——这才是可观测性真正的长期价值。未来，可观测系统将从“被动记录”进化为“主动优化”，这可能是我们下一篇文章的主题。